// Copyright (c) 2000 Just Objects B.V. <just@justobjects.nl>
// Distributable under LGPL license. See terms of license at gnu.org.

package nl.justobjects.pushlet.core;

import java.util.Date;
import java.util.Enumeration;
import java.util.Vector;

/**
 * Dispatcher of Events to Subscribers.
 *
 * @version $Id: Publisher.java,v 1.4 2003/08/15 08:37:40 justb Exp $
 * @author Just van den Broecke - Just Objects &copy;
 **/
public class Publisher {
	private Vector subscribers = new Vector(10);
	private Vector zombieSubscribers = new Vector(5);
	private Subscriber[] arrSubscribers = new Subscriber[5]; // temp cache for notification

	/** Singleton pattern. */
	private static final Publisher instance = new Publisher();

	/** Singleton pattern: private constructor. */
	private Publisher() {
	}

	/** Singleton pattern with lazy construction. */
	public static Publisher getInstance() {
		return instance;
	}

	/** Get number of listening Subscribers.. */
	synchronized public int getSubscriberCount() {
		return subscribers.size();
	}

	/** Get status info. */
	public String getStatus() {
		String statusInfo = null;
		synchronized (subscribers) {
			int nrOfSubscribers = subscribers.size();
			statusInfo = "Publisher: " + nrOfSubscribers + " subscribers (" + PushletSubscriber.getCount() + ")\\n";
			for (int i = 0; i < nrOfSubscribers; i++) {
				statusInfo = statusInfo + (Subscriber) subscribers.elementAt(i) + "\\n";
			}
		}
		return statusInfo;
	}

	/** Add subscriber. */
	synchronized public void join(Subscriber subscriber) {
		if (!subscribers.contains(subscriber)) {
			subscribers.addElement(subscriber);
			log(subscriber.getAddress() + " joined subject " + subscriber.getSubject());

			// Activate event generators on first subscriber
			if (subscribers.size() == 1) {
				EventSourceManager.activate();
			}
		}
	}

	/** Register subscriber for removal. */
	public void leave(Subscriber subscriber) {
		// To avoid deadlock (leaving while notified) we make
		// the subscriber a Zombie first by adding it to the Zombie
		// collection.
		synchronized (zombieSubscribers) {
			if (!zombieSubscribers.contains(subscriber)) {
				zombieSubscribers.addElement(subscriber);
				// p(subscriber.getAddress()+ " leaving subject "+subscriber.getSubject());
			}
		}
	}

	/** Publish event to subscribers matching topic. */
	synchronized public void publish(Event event) {
		// Remove subscribers registered for removal.
		removeZombies();

		int size = subscribers.size();

		// Make new cache if size increased.
		if (size > arrSubscribers.length) {
			arrSubscribers = new Subscriber[size];
		}

		// Copy into local cache
		subscribers.copyInto(arrSubscribers);

		Subscriber nextSubscriber = null;
		String subject = event.getSubject();

		for (int i = size - 1; i >= 0; i--) {
			nextSubscriber =  arrSubscribers[i];

			// Send only if the subscriber's criteria
			// match the event or if probing the client.
			if (nextSubscriber.match(event) || subject.equals("probe")) {
				nextSubscriber.send((Event) event.clone());
				// p("send");
			}

		}

		// Remove subscribers registered for removal (see leave()) during sending.
		removeZombies();
	}

	/** Remove possible subscribers registered for removal. */
	private void removeZombies() {
		synchronized (zombieSubscribers) {

			// Return if no Zombies
			if (zombieSubscribers.size() == 0) {
				return;
			}

			// Remove Zombies from subscriber collection
			Subscriber nextZombie = null;
			for (Enumeration e = zombieSubscribers.elements(); e.hasMoreElements();) {
				nextZombie = (Subscriber) e.nextElement();
				subscribers.removeElement(nextZombie);
				log(nextZombie.getAddress() + " left subject " + nextZombie.getSubject());
			}

			// If no subscribers left, quiet the event generators.
			if (subscribers.size() == 0) {
				EventSourceManager.passivate();
			}

			// Clear zombie collection.
			zombieSubscribers.removeAllElements();
		}
	}

	/** Util: stdout printing. */
	private void log(String s) {
		Log.info("Publisher: " + new Date() + " " + s);
	}
}

/*
 * $Log: Publisher.java,v $
 * Revision 1.4  2003/08/15 08:37:40  justb
 * fix/add Copyright+LGPL file headers and footers
 *
 * Revision 1.3  2003/08/12 08:54:40  justb
 * added getSubscriberCount() and use Log
 *
 * Revision 1.2  2003/05/18 16:15:08  justb
 * support for XML encoded Events
 *
 * Revision 1.1.1.1  2002/09/24 21:02:31  justb
 * import to sourceforge
 *
 * Revision 1.1.1.1  2002/09/20 22:48:18  justb
 * import to SF
 *
 * Revision 1.1.1.1  2002/09/20 14:19:04  justb
 * first import into SF
 *
 * Revision 1.3  2002/04/15 20:42:41  just
 * reformatting and renaming GuardedQueue to BlockingQueue
 *
 * Revision 1.2  2000/08/21 20:48:29  just
 * added CVS log and id tags plus copyrights
 *
 *
 */
